package com.fulu.game.bi.mysql.config;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.ExceptionSorter;
import com.fulu.game.bi.mysql.properties.MultipleDatabaseProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;

import javax.sql.DataSource;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;


@Slf4j
@Configuration
@EnableConfigurationProperties(
        {MultipleDatabaseProperties.class})
@MapperScan(basePackages = {"com.fulu.game.bi.hbase"},
        sqlSessionFactoryRef = "hBaseSqlSessionFactory")
public class HbaseConfig extends AbstractDbConfig {

    @Autowired
    private MultipleDatabaseProperties multipleDatabaseProperties;

    @Bean(name = "hBaseDataSource")
    public DataSource dataSource() {
        DataSourceProperties dbProperties = multipleDatabaseProperties.getHBaseDb();
        DruidDataSource dataSource = new DruidDataSource ();
        dataSource.setDriverClassName (dbProperties.getDriverClassName ());
        dataSource.setUrl (dbProperties.getUrl ());
        dataSource.setUsername (dbProperties.getUsername ());
        dataSource.setPassword (dbProperties.getPassword ());
        dataSource.setInitialSize (10);
        dataSource.setMinIdle (10);
        dataSource.setMaxActive (300);
        dataSource.setMaxWait (30000);
        dataSource.setRemoveAbandoned (true);
        dataSource.setRemoveAbandonedTimeout (1800);
        dataSource.setLogAbandoned (false);
        dataSource.setTimeBetweenEvictionRunsMillis (60000);
        dataSource.setMinEvictableIdleTimeMillis (300000);
        dataSource.setValidationQuery ("SELECT 1");
        dataSource.setTestWhileIdle (true);
        dataSource.setTestOnBorrow (false);
        dataSource.setTestOnReturn (false);
        dataSource.setPoolPreparedStatements (true);
        dataSource.setMaxPoolPreparedStatementPerConnectionSize (500);
        dataSource.setKeepAlive (true);
        dataSource.setExceptionSorter (new PhoenixExceptionSorter());
        try
        {
            dataSource.setFilters ("druidWebStatFilter,stat-filter,log-filter,stat");
        }
        catch (Exception e)
        {
            log.error ("设置过滤器异常:", e);
        }
        return dataSource;
    }

    /**
     * 当网络断开或者数据库服务器Crash时，连接池里面会存在“不可用连接”，连接池需要
     * 一种机制剔除这些“不可用连接”。在Druid和JBoss连接池中，剔除“不可用连接”的机
     * 制称为ExceptionSorter，实现的原理是根据异常类型/Code/Reason/Message来识
     * 别“不可用连接”。没有类似ExceptionSorter的连接池，在数据库重启或者网络中断之
     * 后，不能恢复工作，所以ExceptionSorter是连接池是否稳定的重要标志。
     * @author mgguan   2019-10-17 15:00:00
     */
    class PhoenixExceptionSorter implements ExceptionSorter {
        @Override
        public boolean isExceptionFatal(SQLException e) {
            if (e.getMessage().contains("Connection is null or closed")) {
                log.error("剔除phoenix不可用的连接", e);
                return true;
            }
            return false;
        }

        @Override
        public void configFromProperties(Properties properties) {
        }
    }

    @Bean(name = "hBaseTransactionManager")
    public DataSourceTransactionManager hBaseTransactionManager(@Qualifier("hBaseDataSource") DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }

    @Bean(name = "hBaseSqlSessionTemplate")
    public SqlSessionTemplate hBaseSqlSessionTemplate(@Qualifier("hBaseSqlSessionFactory") SqlSessionFactory sqlSessionFactory) throws Exception {
        return new SqlSessionTemplate(sqlSessionFactory);
    }

    @Bean(name = "hBaseSqlSessionFactory")
    public SqlSessionFactory hbasePhoenixSqlSessionFactory(@Qualifier("hBaseDataSource") DataSource dataSource)
            throws Exception {
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(dataSource);
        List<Resource> resources = new ArrayList<>();
        PathMatchingResourcePatternResolver pathMatchingResourcePatternResolver = new PathMatchingResourcePatternResolver();
        resources.addAll(Arrays.asList(pathMatchingResourcePatternResolver.getResources("classpath:META-INF/hbase/*.xml")));
        bean.setMapperLocations(resources.toArray(new Resource[resources.size()]));
        bean.getObject().getConfiguration().setMapUnderscoreToCamelCase(true);
        return bean.getObject();
    }
}
